package com.shujia.flink.demo

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{MapState, MapStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object Demo2CountyCount {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 1、读取电信数据
      *
      */

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id", "asdasdasd")


    //创建消费者
    val flinkKafkaConsumer = new FlinkKafkaConsumer[String](
      "dianxin",
      new SimpleStringSchema(),
      properties
    )

    //指定读取数据的位置
    flinkKafkaConsumer.setStartFromEarliest() // 尽可能从最早的记录开始


    //读取kafka
    val dianxinDS: DataStream[String] = env.addSource(flinkKafkaConsumer)

    //过滤脏数据
    val filterDS: DataStream[String] = dianxinDS.filter(_.split(",")(3) != "\\N")

    //取出手机号和区县编号
    val kvDS: DataStream[(String, String)] = filterDS.map(line => {
      val split: Array[String] = line.split(",")
      val mdn: String = split(0)
      val countyId: String = split(3)

      val date = new Date()
      val format = new SimpleDateFormat("yyyyMMdd")
      val day: String = format.format(date)

      val key: String = countyId + "_" + day

      (mdn, key)
    })


    //按照区县keyby
    val keyByDS: KeyedStream[(String, String), String] = kvDS.keyBy(_._2)

    //实时统计区县的人流量
    val countDS: DataStream[(String, String, Long)] = keyByDS.process(new MyProcess)


    //将结果保存到mysql中
    countDS.addSink(new MysqlSink)


    env.execute()

  }

}

class MyProcess extends KeyedProcessFunction[String, (String, String), (String, String, Long)] {

  var mapState: MapState[String, Int] = _
  var valueState: ValueState[Long] = _

  override def open(parameters: Configuration): Unit = {
    val context: RuntimeContext = getRuntimeContext

    val mapStateDesc = new MapStateDescriptor[String, Int]("map", classOf[String], classOf[Int])

    //key 用于保存手机号，去重， value没有用
    mapState = context.getMapState(mapStateDesc)


    val valueStateDesc = new ValueStateDescriptor[Long]("count", classOf[Long])

    //保存每一个区县的人流量
    valueState = context.getState(valueStateDesc)

  }

  override def processElement(value: (String, String), ctx: KeyedProcessFunction[String, (String, String), (String, String, Long)]#Context, out: Collector[(String, String, Long)]): Unit = {

    /**
      * 每天0点清空之前的状态
      *
      */
    //获取当前时间
    val ts: Long = System.currentTimeMillis()
    if (ts % (24 * 60 * 60 * 1000) == 0) {
      mapState.clear()
      valueState.clear()
    }


    val mdn: String = value._1
    val counttyIdAndDay: String = value._2

    val split: Array[String] = counttyIdAndDay.split("_")
    val countyId: String = split(0)
    val day: String = split(1)


    //判断手机号之前是否已经存在
    if (!mapState.contains(mdn)) {
      //将当前手机号保存到mapState中
      mapState.put(mdn, 1)

      //在之前的结果上加一
      val currCount: Long = valueState.value() + 1

      //更新人流量
      valueState.update(currCount)

      //返回最新的人流量
      out.collect((countyId, day, currCount))
    }


  }
}


class MysqlSink extends RichSinkFunction[(String, String, Long)] {

  var con: Connection = _

  override def open(parameters: Configuration): Unit = {
    //使jdbc读取mysql中的数据
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //2、创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")

  }

  override def close(): Unit = {
    con.close()
  }

  override def invoke(value: (String, String, Long), context: SinkFunction.Context[_]): Unit = {
    val countyId: String = value._1
    val day: String = value._2
    val count: Long = value._3


    val stat: PreparedStatement = con.prepareStatement("insert into county_count(countyId,day,c) values(?,?,?) ON DUPLICATE KEY UPDATE countyId=?,day=?,c=?")

    stat.setString(1, countyId)
    stat.setString(2, day)
    stat.setLong(3, count)

    stat.setString(4, countyId)
    stat.setString(5, day)
    stat.setLong(6, count)

    stat.executeUpdate()


  }
}
